package com.ruoyi.quartz.task;


import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.text.DecimalFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

@Component("kafka")
@Slf4j
public class O1KafkaBatch {

    @Value("${kafka.topic.temperature}")
    private String topic;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private final Random random = new Random();
    private final DecimalFormat df = new DecimalFormat("#.##");
    private final AtomicInteger counter = new AtomicInteger(0);
    private final DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss.SSS");
    public void batchProduceMessage(String params)
    {
        int batchSize = 5000; // 每批消息数量
        for (int i = 0; i < batchSize; i++) {
            String message = buildTemperatureMessage();
            ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, message);
            send.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable ex) {
                    log.error("消息发送失败: message={}, error={}", message, ex.getMessage());
                }

                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("消息发送成功: topic={}, offset={}", topic, result.getRecordMetadata().offset());

                }
            });
        }
        System.out.printf("[%s] 生产者发送批次: %d | 总消息量: %d\n",
                LocalDateTime.now().format(timeFormatter),
                batchSize,
                counter.addAndGet(batchSize));
    }


    private String buildTemperatureMessage() {
        double temperature = 15 + 20 * random.nextDouble();
        return String.format("{\"time\":\"%s\",\"temp\":%s,\"sensor_id\":%d}",
                LocalDateTime.now().format(timeFormatter),
                df.format(temperature),
                random.nextInt(10));
    }
}
